package org.jlambda.util;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.jlambda.Context;
import org.jlambda.Context.BackgroundExecutor;
import org.jlambda.exceptions.LamdaException;
import org.jlambda.functions.Fun1;
import org.jlambda.functions.Fun2;
import org.jlambda.list.List;
import org.jlambda.listFunctions.ListFun5;
import org.jlambda.tuples.Tuple2;

/**
 * Contains functions related to concurrent execution.
 * 
 * @author Chris
 *
 */
final public class ConcurrentUtils {

	/**
	 * Calls the function on each element of the list, creating a new list.
	 * The produced list is not in the same order as the original list.
	 * 
	 * When each item is requested from the list it will apply at most look_ahead
	 * applications in the background. These will be in varying block sizes upto 
	 * max_group_size.  Since these tasks may be processed in any order in the
	 * background the resulting order is not gauranteed.  The defaults of (5,20) are
	 * applied.  The first group will be one, then two.. 5,1,2 etc until at least
	 * 20 have been processed.
	 * 
	 * The results of each background group processed are added to an internal
	 * blocking queue.  As such the next item in the iterator may block indefinitely,
	 * the default value is 10 minutes but a tuple of (Long, TimeUnit) can be used, 
	 * null indicating default.
	 * 
	 * Not memoized by default
	 * 
	 * An optional BackgroundExecutor can be supplied.
	 * 
	 * mapBg BackgroundExecutor (Long,TimeUnit) (Integer max_group_size,Integer look_ahead) function ITERABLE
	 * 
	 */
	public static class MapBg<A, T> extends ListFun5<BackgroundExecutor, Tuple2<Long,TimeUnit>, Tuple2<Integer,Integer>, Fun1<A,T>, Iterable<A>, T>{
		Iterator<A> itr = null;
		Fun1<A, T> f = null;
		
		boolean allprocessed = false;
		long sourceSize = 0;
		long retrieved = 0;
		
		// 10 minutes
		long timeunit = 600;
		TimeUnit timeU = TimeUnit.SECONDS;
		
		// groups, min look ahead 20 items
		int max_group_size = 5;
		int look_ahead = 20;
		
		BlockingQueue<ArrayList<T>> queue = null;
		
		final Fun2<Fun1<A,T>,ArrayList<A>, Void> callOver = new Fun2<Fun1<A,T>,ArrayList<A>, Void>(){
			@Override
			public Void apply(Fun1<A,T> f, ArrayList<A> p1) {
				ArrayList<T> results = new ArrayList<T>();
				for (A a : p1) {
					results.add(f.apply(a));
				}
				// add to queue
				try {
					queue.put(results);
				} catch (InterruptedException e) {
					// not use
				}
				return null;
			}
		};
		private BackgroundExecutor exec = null;
		
		protected ArrayList<T> nextResults = null;
		
		/**
		 * fire off look ahead, poll for results 
		 */
		public T getNext() {
//			if (allprocessed) {
//				EOL();
//				return null;
//			}
			int group = 1;
			int counted = 0;
			while(counted < look_ahead && !allprocessed) {
				ArrayList<A> res = new ArrayList<A>();
				for (int i = 0; i < group && !allprocessed; i++) {
					if (itr.hasNext()) {
						res.add(itr.next());
						counted ++;
					} else {
						allprocessed = true; // may only happen for finite lists
					}
				}
				// process the group
				if (res.size() != 0) {
					callOver.applyBg(exec, f, res);
				}
				group++;
			}
			sourceSize += counted;
			
			// if items in nextResults carry on
			if (nextResults.size() != 0){
				retrieved ++;
				return nextResults.remove( nextResults.size() - 1 );
			}
			
			if (retrieved != sourceSize) {
				// get next result set
				try {
					nextResults = queue.poll(timeunit, timeU);
				} catch (InterruptedException e) {
					throw new LamdaException("Timeout expired, no items in list", e);
				}
				
				// correct because there must be at least one in the list
				// if items in nextResults carry on
				if (nextResults.size() != 0){
					retrieved ++;
					return nextResults.remove( nextResults.size() - 1 );
				}
			}

			// if we get here, we've got no more items to retrieve
			EOL();
			return null;
		}

		@Override
		protected void create(List<T> list, BackgroundExecutor ex, Tuple2<Long,TimeUnit> times, Tuple2<Integer,Integer> groups, Fun1<A, T> p1, Iterable<A> p2) {
			f = p1;
			itr = p2.iterator();
			if (ex == null){
				exec  = Context.getExecutor();
			} else {
				exec = ex;
			}
			
			if (times != null){
				timeunit = times.get1();
				timeU = times.get2();
			}
			if (groups != null) {
				max_group_size = groups.get1();
				look_ahead = groups.get2();
			}
			queue = new LinkedBlockingQueue<ArrayList<T>>();
			nextResults = new ArrayList<T>();
			
			setMemoized(false);
		}
		
	}

	public static final <A,T> Iterable<T> mapBg(BackgroundExecutor ex, Tuple2<Long,TimeUnit> times, Tuple2<Integer,Integer> groups, Fun1<A, T> p1, Iterable<A> p2) {
		MapBg<A,T> map = new MapBg<A,T>();
		return map.apply(ex, times, groups, p1, p2);
	}
	
	public static final <A,T> Iterable<T> mapBg(Tuple2<Long,TimeUnit> times, Tuple2<Integer,Integer> groups, Fun1<A, T> p1, Iterable<A> p2) {
		return mapBg(null, times, groups, p1, p2);
	}
	
	public static final <A,T> Iterable<T> mapBg(Tuple2<Integer,Integer> groups, Fun1<A, T> p1, Iterable<A> p2) {
		return mapBg(null, null, groups, p1, p2);
	}
	
	public static final <A,T> Iterable<T> mapBg(Fun1<A, T> p1, Iterable<A> p2) {
		return mapBg(null, null, null, p1, p2);
	}
	
}
